package org.com.blbl.activeProvince;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class ActiveProvinceReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    //provinceCountMap 是一个 HashMap，用于存储每个省份及其对应的活动数量。键是省份名称，值是活跃省份数量。
    private Map<String, Integer> provinceCountMap = new HashMap<>();

    /*

    1.reduce 方法是核心逻辑，负责处理每个输入的键值对。
    2.key 是省份名称，values 是与该省份相关的活动数量的集合。
    3.sum 变量用于累加该省份的活动数量。
    4.遍历 values，将每个 IntWritable 值累加到 sum 中。
    5.最后，将省份名称和其总活动数量存入 provinceCountMap 中。

     */
    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        provinceCountMap.put(key.toString(), sum);
    }

    /*
    1.cleanup 方法在所有的 reduce 调用完成后执行，通常用于输出最终结果。
    2.provinceCountMap.entrySet().stream() 创建一个流，允许对 provinceCountMap 的条目进行操作。
    3.sorted(Map.Entry.<String, Integer>comparingByValue().reversed()) 将条目按值（活动数量）进行降序排序。
    4.limit(5) 限制输出前5个条目，即活动数量最多的5个省份。
    5.forEach(entry -> {...}) 遍历这些条目，并将它们写入上下文中，输出省份名称和活动数量。
    6.在写入过程中，捕获并处理可能的异常。
     */
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        provinceCountMap.entrySet().stream()
                .sorted(Map.Entry.<String, Integer>comparingByValue().reversed())
                .limit(5)
                .forEach(entry -> {
                    try {
                        context.write(new Text(entry.getKey()), new IntWritable(entry.getValue()));
                    } catch (IOException | InterruptedException e) {
                        e.printStackTrace();
                    }
                });
    }
}
